package producer

import (
	"encoding/json"
	"log"

	"github.com/Shopify/sarama"

	"kafka_go/common"
)

type Producer struct {
	sarama.SyncProducer
	topic string
}

func NewProducer(topic string) *Producer {
	pro := common.NewSyncProducer()
	e := &Producer{
		pro,
		topic,
	}
	return e
}

func (e *Producer) SendMsg(data *common.KafkaMsg) (err error) {
	bytes, err := json.Marshal(data)
	if err != nil {
		log.Println("marshal data failed", "err is ", err.Error())
		return err
	}
	msg := &sarama.ProducerMessage{
		Value: sarama.ByteEncoder(bytes),
		Topic: e.topic,
	}
	_, _, err = e.SendMessage(msg)
	return err
}
